package polimi.reds.optimizer;

import java.io.IOException;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.logging.Logger;

import polimi.reds.Filter;
import polimi.reds.Message;
import polimi.reds.MessageID;
import polimi.reds.NodeDescriptor;
import polimi.reds.broker.overlay.GenericOverlay;
import polimi.reds.broker.overlay.Link;
import polimi.reds.broker.overlay.NotConnectedException;
import polimi.reds.broker.overlay.NotRunningException;
import polimi.reds.broker.overlay.TopologyManager;
import polimi.reds.broker.overlay.Transport;
import polimi.reds.broker.routing.SubscriptionTable;

public class SwapCapableGenericOverlay extends GenericOverlay {

	protected Logger logger = Logger.getLogger("polimi.reds.optimizer.SwapCapableGenericOverlay");
	private TrafficLock trafficLock;
	private Queue<OverlayAction> queue = new ConcurrentLinkedQueue<OverlayAction>();
	private QueueActionPerformer queueThread;
	private HashMap<MessageID,Boolean> cache = new HashMap<MessageID,Boolean>(); // TODO LIMIT THIS CACHE!
	private TopologyManager topoMgr;
	private SubscriptionTable table;
	
	public SwapCapableGenericOverlay(TrafficLock trafficLock, TopologyManager topoMgr, Transport tr, SubscriptionTable table) {
		super(topoMgr, tr);
		this.trafficLock = trafficLock;
		this.topoMgr = topoMgr;
		this.table = table;
	}
	
	
	@Override
	public void send(String subject, Serializable packet,
			NodeDescriptor recipient) throws IOException,
			NotConnectedException, NotRunningException {
		
		logger.fine(getNodeDescriptor() + ": processing SEND (" + subject + ") " + packet + " to " + recipient
				+ " [ " + getNeighbouringInfo() + "]");
		
		super.send(subject, packet, recipient);
	}
		/*
		if (!isRunning()) {
			throw new NotRunningException();
		}
		
		if (!isNeighborOf(recipient)) {
			throw new NotConnectedException();
		}
		
		if (recipient.isClient() && packet instanceof Message) {
			if (checkDuplicate(new MessageIDWithDestination(((Message) packet).getID(),recipient))) {
				return;
			}
		}
		
		if (isQueueExceptionSubject(subject)) {
			super.send(subject, packet, recipient);
		} else {
			System.out.println(getNodeDescriptor() + ": queuing SEND message " + packet + " to " + recipient);
			queue.add(new DeferredSendAction(subject, packet, recipient));
		}
	}
	
	private void superSend(String subject, Serializable packet,
			NodeDescriptor recipient) throws IOException,
			NotConnectedException, NotRunningException {
		super.send(subject, packet, recipient);
	}*/

	@Override
	public void start() {
		super.start();
		if (queueThread == null || !queueThread.isAlive()) {
			queueThread = new QueueActionPerformer();
			queueThread.setDaemon(true);
			queueThread.start();
		}
	}
	
	@Override
	public void notifyDataArrived(String subject, Link source, Serializable data) {
		
		/*if (data instanceof Message) {
			Message msg = (Message) data;
			if (checkDuplicate(msg.getID())) {
				System.out.println(getNodeDescriptor() + ": *ignoring duplicate* RECEIVE (" + subject + ") message " + data + " from " + topoMgr.getNeighborFor(source)
						+ " [ " + getNeighbouringInfo() + "]");
				return;
			}
		}*/
		
		NodeDescriptor from = topoMgr.getNeighborFor(source);
		
		synchronized (trafficLock) {

			if (!trafficLock.isLocked() && !queue.isEmpty()) {
				logger.fine(getNodeDescriptor() + ": queuing/synchronizing RECEIVE (" + subject + ") message " + data + " from " + topoMgr.getNeighborFor(source)
						+ " [ " + getNeighbouringInfo() + "]");
				trafficLock.reportQueueActivity();
				queue.add(new DeferredNotifyDataArrivedAction(subject, from, data));
			} else if (!trafficLock.isLocked() || from.equals(trafficLock.getN0()) || from.equals(trafficLock.getN1()) || from.equals(trafficLock.getN2())) {
				synchronized(topoMgr) {
					logger.fine(getNodeDescriptor() + ": processing RECEIVE (" + subject + ") message " + data + " from " + topoMgr.getNeighborFor(source)
							+ " [ " + getNeighbouringInfo() + "]");
				}
				super.notifyDataArrived(subject, source, data);
			} else {
				logger.fine(getNodeDescriptor() + ": queuing RECEIVE (" + subject + ") message " + data + " from " + topoMgr.getNeighborFor(source)
						+ " [ " + getNeighbouringInfo() + "]");
				trafficLock.reportQueueActivity();
				queue.add(new DeferredNotifyDataArrivedAction(subject, from, data));
			}
		}
		
		/*
		if (isQueueExceptionSubject(subject)) {
			super.notifyDataArrived(subject, source, data);
		} else {
			System.out.println(getNodeDescriptor() + ": queuing RECEIVE message " + data + " from " + topoMgr.getNeighborFor(source));
			queue.add(new DeferredNotifyDataArrivedAction(subject, topoMgr.getNeighborFor(source), data));
		}*/
	}
	
	private void superNotifyDataArrived(String subject, Link source, Serializable data) {
		super.notifyDataArrived(subject, source, data);
	}

	private class QueueActionPerformer extends Thread {
		public void run() {
			while (isRunning()) {
				try {
					Thread.sleep(10);
				} catch (InterruptedException e) {
				}
				
				if (!queue.isEmpty()) {
					synchronized(trafficLock) {
						while (!trafficLock.isLocked() && !queue.isEmpty()) {
							trafficLock.reportQueueActivity();
							queue.remove().doAction();
						}
					}
				}
			}
		}
	}
	
	public boolean checkDuplicate(MessageID mid) {
		if (cache.get(mid) != null) {
			return true;
		} else {
			cache.put(mid, true);
			return false;
		}
	}
	
	/*public boolean checkDuplicate(MessageIDWithDestination mid) {
		
		if (cache.get(mid) != null) {
			return true;
		} else {
			cache.put(mid, true);
			return false;
		}
	}*/
	
	abstract class OverlayAction {
		abstract public void doAction();
	}
	
	/*
	private class DeferredSendAction extends OverlayAction {
		private String subject;
		private Serializable packet;
		private NodeDescriptor recipient;
		
		public DeferredSendAction(String subject, Serializable packet,
				NodeDescriptor recipient) {
			this.subject = subject;
			this.packet = packet;
			this.recipient = recipient;
		}
		
		public void doAction() {
			try {
				System.out.println(getNodeDescriptor() + ": unqueuing SEND message " + packet + " to " + recipient);
				superSend(subject, packet, recipient);
			} catch (Exception e) { 
				e.printStackTrace();
			}
		}
	}*/
	
	private class DeferredNotifyDataArrivedAction extends OverlayAction {
		private String subject;
		private NodeDescriptor source;
		private Serializable data;
		
		public DeferredNotifyDataArrivedAction(String subject, NodeDescriptor source, Serializable data) {
			this.subject = subject;
			this.source = source;
			this.data = data;
		}
		
		public void doAction() {
			
			Link sourceLink = topoMgr.getLinkFor(source);
			
			if (sourceLink != null) {
				logger.fine(getNodeDescriptor() + ": unqueuing RECEIVE (" + subject + ")message " + data + " from " + source
						+ " [ " + getNeighbouringInfo() + "]");
				superNotifyDataArrived(subject, sourceLink, data); // FIXME
				logger.fine(getNodeDescriptor() + ": notified RECEIVE (" + subject + ")message " + data + " from " + source + " [ " + getNeighbouringInfo() + "]");
			} else {
				try {
					throw new NotConnectedException();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	private class MessageIDWithDestination {
		
		private MessageID id;
		private NodeDescriptor dest;
		
		
		public MessageIDWithDestination(MessageID id, NodeDescriptor dest) {
			this.id = id;
			this.dest = dest;
		}
		
		public boolean equals(Object o) {
			if (o instanceof MessageIDWithDestination) {
				MessageIDWithDestination mid = (MessageIDWithDestination) o;
				return mid.id.equals(id) && mid.dest.equals(dest);
			}
			
			return false;
		}
		
		public String toString() {
			return id.toString() + " " + dest.getID();
		}
		
		public int hashCode() {
			return toString().hashCode();
		}
	}
	
	private boolean isQueueExceptionSubject(String subject) {
		return subject.equals(OptimizerStrategy.REMOTE_SWAP_N1) || 
			subject.equals(OptimizerStrategy.REMOTE_SWAP_N2) || 
			subject.equals(OptimizerStrategy.REMOTE_ACK);
	}
	
	private String getNeighbouringInfo() {
		String s="";
		for (NodeDescriptor n:getNeighbors()) {
			s += n + "{ ";
			for (Object f:table.getAllFilters(n)) {
				s += f + " ";
			}
			
			s += " } ";
		}
		return s;
	}

}
